package com.lmm.rpc.core.proxy.invocation;

import com.lmm.rpc.core.client.NettyClient;
import com.lmm.rpc.core.client.protocol.MessageHeader;
import com.lmm.rpc.core.client.protocol.MessageProtocol;
import com.lmm.rpc.core.client.protocol.RequestMetadata;
import com.lmm.rpc.core.cluster.*;
import com.lmm.rpc.core.common.reponse.DefaultRepose;
import com.lmm.rpc.core.common.request.DefaultRequest;
import com.lmm.rpc.core.exception.RpcException;
import io.netty.channel.Channel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.impl.AbandonedConfig;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.ConnectException;

/**
 * @title: ClientStubInvocationHandler
 * @Description: 客户端代理增强增强类
 * @Author Lmm
 * @Date: 2022/10/8 12:53
 * @Version 1.0
 */
@Data
@Slf4j
public class ClientStubInvocationHandler implements InvocationHandler {
    //节点
    private ICluster cluster;
    //负载
    private ILoadBalancer iLoadBalancer;
    //核心cpu数
    private static int cores = Runtime.getRuntime().availableProcessors();

    //环境
    private String env="dev";

    //client端service接口
    private String serviceInterface;

    //2种方式
    // 方式1：zk管理的动态集群,格式192.168.3.253:6666
    private String zkPath;
    // 方式2：指定的server列表，逗号分隔，#分隔权重,格式192.168.3.253:6666#10,192.168.3.253:6667#10
    private String serverIpPorts;

    // 同步还是异步,默认同步。 todo
    private boolean async = false;
    // 是否泛化调用  todo
    private boolean generic = false;

    //连接超时
    public static final int DEFAULT_CONNTIMEOUT = 5*1000;
    //读取超时
    public static final int DEFUAL_READTIMEOUT = 30*1000;
    //请求体最大长度
    public static final int DEFUAL_MAXLENGTH = 10 * 1024 * 1024;

    //重试
    private boolean retryRequest = true;
    private int retryTimes = 3;


    //抛弃策略
    private AbandonedConfig abandonedConfig;
    private boolean removeAbandonedOnBorrow = true;
    private boolean removeAbandonedOnMaintenance = true;
    private int removeAbandonedTimeout = 30;

    //公钥私钥  支持ssl加密传输 todo
    private String privateKey;
    private String publicKey;

    //是否开启链路监听
    private boolean cat=false;

    //池配置
    private GenericObjectPoolConfig genericObjectPoolConfig;
    //池相关配置
    //最大闲置数
    private int maxIdle=20;
    //最小闲置数量
    private int minIdle=10;
    //最大连接数
    private int maxTotal=50;
    //最大等待时间
    private long maxWaitMillis = 30 * 1000;
    private boolean lifo = true;
    private boolean fairness = false;

    //多长时间运行一次
    private long timeBetweenEvictionRunsMillis = 3 * 60 * 1000;
    private long minEvictableIdleTimeMillis = 5 * 60 * 1000;
    //对象空闲多久后逐出, 当空闲时间>该值 且 空闲连接>最大空闲数 时直接逐出,
    //不再根据MinEvictableIdleTimeMillis判断  (默认逐出策略)
    private long softMinEvictableIdleTimeMillis = 10 * 60 * 1000;
    private int numTestsPerEvictionRun = 20;
    private boolean testOnCreate = false;
    private boolean testOnBorrow = false;
    private boolean testOnReturn = false;
    private boolean testWhileIdle = true;
    private GenericObjectPoolConfig getGenericObjectPoolConfig() {
        GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig ();
        genericObjectPoolConfig.setMaxTotal ( getMaxTotal () );
        genericObjectPoolConfig.setMinIdle ( getMinIdle () );
        genericObjectPoolConfig.setMaxIdle ( maxIdle );
        genericObjectPoolConfig.setMaxWaitMillis ( getMaxWaitMillis () );
        genericObjectPoolConfig.setLifo ( isLifo () );
        genericObjectPoolConfig.setFairness ( isFairness () );
        genericObjectPoolConfig.setMinEvictableIdleTimeMillis ( getMinEvictableIdleTimeMillis () );
        genericObjectPoolConfig.setSoftMinEvictableIdleTimeMillis ( getSoftMinEvictableIdleTimeMillis () );
        genericObjectPoolConfig.setNumTestsPerEvictionRun ( getNumTestsPerEvictionRun () );
        genericObjectPoolConfig.setTimeBetweenEvictionRunsMillis ( getTimeBetweenEvictionRunsMillis () );
        genericObjectPoolConfig.setTestOnCreate ( isTestOnCreate () );
        genericObjectPoolConfig.setTestOnBorrow ( isTestOnBorrow () );
        genericObjectPoolConfig.setTestOnReturn ( isTestOnReturn () );
        genericObjectPoolConfig.setTestWhileIdle ( isTestWhileIdle () );
        return genericObjectPoolConfig;
    }

    private AbandonedConfig getAbandonedConfig() {
        AbandonedConfig abandonedConfig = new AbandonedConfig ();
        abandonedConfig.setRemoveAbandonedOnBorrow ( isRemoveAbandonedOnBorrow () );
        abandonedConfig.setRemoveAbandonedOnMaintenance ( isRemoveAbandonedOnMaintenance () );
        abandonedConfig.setRemoveAbandonedTimeout ( getRemoveAbandonedTimeout () );
        return abandonedConfig;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

        //排除toString、hashCode、equals 等函数
        String methodName = method.getName ();
        if ("toString".equals ( methodName ) && args.length == 0) {
            return this.toString ();
        }
        if ("hashCode".equals ( methodName ) && args.length == 0) {
            return this.hashCode ();
        }
        if ("equals".equals ( methodName ) && args.length == 1) {
            return this.equals ( args[0] );
        }

            Channel channel = null;
            int tryTime = 0;
            while (tryTime++ < retryTimes){
                ServerObject<Channel> objectPool = cluster.getObjectForRemote();
                if (objectPool == null)
                    throw new RuntimeException("no server list to use :"+serviceInterface);
                GenericObjectPool<Channel> channelGenericObjectPool = objectPool.getGenericObjectPool();

                try{
                    long before = System.currentTimeMillis ();
                    channel = channelGenericObjectPool.borrowObject();
                    long after = System.currentTimeMillis ();
                    log.debug("get channel from pool in {}ms,serviceInterface"+serviceInterface+"methodName:"+method.getName(),after - before);
                }catch (Exception e){
                    if (channel != null){
                        channelGenericObjectPool.returnObject(channel);
                    }
                    log.error("borrow channel from pool error,serviceInterface:"+serviceInterface);
                    throw new RpcException("borrow channel from pool fail! reason {}",e);
                    //todo 是否需要把节点信息打印出来
                }

                NettyClient nettyClient = cluster.getNettyClient();
                MessageProtocol<DefaultRequest> messageProtocol = new MessageProtocol<>();
                // 设置请求头
                messageProtocol.setHeader(MessageHeader.build("HESSIAN"));
                // 设置请求体
                DefaultRequest request = new DefaultRequest();
                request.setInterfaceName(serviceInterface);
                request.setMethodName(method.getName());
                request.setParameterTypes(method.getParameterTypes());
                request.setArguments(args);
                messageProtocol.setBody(request);

                MessageProtocol<DefaultRepose> reposeMessageProtocol = null;
                try {
                    reposeMessageProtocol = nettyClient.sendRequest(RequestMetadata.builder().protocol(messageProtocol).build(), channel);
                    channelGenericObjectPool.returnObject(channel);
                    if (reposeMessageProtocol == null){
                        log.error("request timeout!");
                        throw new ConnectException("请求超时!");
                    }
                    return reposeMessageProtocol.getBody().getData();
                }catch (Exception e){
                    Throwable cause = (e.getCause () == null) ? e : e.getCause ();

                    if (cause.getCause() != null && cause.getCause() instanceof ConnectException){
                        log.warn("the server 【{}】 maybe shutdown , retry other one again! serverName:{},method:{}",objectPool.getRemoteServer(),serviceInterface,method.getName());
                        if (channel != null)
                            channelGenericObjectPool.returnObject(channel);
                        if (retryRequest){
                            continue;
                        }
                        throw new RpcException("serviceName:"+serviceInterface+"do not work!");
                    }
                    log.error("invoke server error,server ip -【{}】,port -【{}】--serverName【{}】,method：【{}】",objectPool.getRemoteServer ().getIp (), objectPool.getRemoteServer ().getPort (),serviceInterface,method.getName());
                    throw cause;
                }
            }
            throw new RpcException("retry 3 times , do not work! please check serviceName:"+serviceInterface);
    }

    /**
     * 检查必要参数是否初始化
     * cluster
     * iLoadBalancer
     * zkPath
     * serverIpPorts
     * 如果没有则抛出异常
     * 注册中心拉取服务
     * */
    public void afterProperties(){
        if (serviceInterface == null){
            throw new IllegalArgumentException("serviceInterface can't be bull! ");
        }
        if (zkPath == null && serverIpPorts == null ){
            throw new IllegalArgumentException("zkPath or serverIpPorts can't be bull at least once! ");
        }
        //初始池的配置
        genericObjectPoolConfig = getGenericObjectPoolConfig();
        abandonedConfig = getAbandonedConfig();

        //初始化注册中心 拉取服务
        if (serverIpPorts != null && serverIpPorts.trim().length() > 0){
            cluster = new DirectClusterImpl(serverIpPorts,iLoadBalancer,serviceInterface,false,DEFAULT_CONNTIMEOUT,DEFUAL_READTIMEOUT,genericObjectPoolConfig,abandonedConfig);
        }else {
            cluster = new ZKClusterImpl(zkPath,iLoadBalancer,serviceInterface,env,false,DEFAULT_CONNTIMEOUT,DEFUAL_READTIMEOUT,genericObjectPoolConfig,abandonedConfig);
        }
    }
}
